package ds_industry_2025.ds.ds_06.T4


import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{Row, SparkSession, functions}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.DoubleType

import java.util.Properties

/*
    2、根据第1小题的结果，对其进行聚合，其中对sku_id进行one-hot转换，将其转换为以下格式矩阵：第一列为用户id，其余列名为商品id，按
    照用户id进行升序排序，展示矩阵第一行前5列数据，将结果截图粘贴至客户端桌面【Release\任务C提交结果.docx】中对应的任务序号下。
字段	类型	中文含义	备注
user_id	double	客户key
sku_id0	double	用户是否购买过商品1	若用户购买过该商品，则值为1，否则为0
sku_id1	double	用户是否购买过商品2	若用户购买过该商品，则值为1，否则为0
sku_id2	double	用户是否购买过商品3	若用户购买过该商品，则值为1，否则为0
.....
结果格式如下：
---------------第一行前5列结果展示为---------------
0.0,1.0,0.0,0.0,0.0
 */
object t2 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("特征工程02")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()


    val conn = new Properties()
    conn.setProperty("user", "root")
    conn.setProperty("password", "123456")
    conn.setProperty("driver", "com.mysql.jdbc.Driver")

    //  导入隐式转换
    import spark.implicits._

    //  todo 读取需要用到的表的数据
    val sku_info = spark.read
      .jdbc("jdbc:mysql://192.168.40.110:3306/shtd_store?useSSL=false", "sku_info", conn)

    val order_info = spark.read
      .jdbc("jdbc:mysql://192.168.40.110:3306/shtd_store?useSSL=false", "order_info", conn)

    val detail_info = spark.read
      .jdbc("jdbc:mysql://192.168.40.110:3306/shtd_store?useSSL=false", "order_detail", conn)

    val user_info = spark.read
      .jdbc("jdbc:mysql://192.168.40.110:3306/shtd_store?useSSL=false", "user_info", conn)


    //  todo 剔除订单信息表(order_info)和订单详细表(order_detail)中的user_id和sku_id
    //   不存在现有维度表(sku_info和user_info)中的数据
    val user_id = user_info.select("id").withColumnRenamed("id", "id_1").distinct()
    val sku_id = sku_info.select("id").withColumnRenamed("id", "id_1").distinct()
    val order = order_info.join(user_id, user_id("id_1") === order_info("user_id"))
    val detail = detail_info.join(sku_id, sku_id("id_1") === detail_info("sku_id")).drop("id")


    //  todo 根据user_id和sku_id升序排列的方式使用排序函数转化为题目要求的格式
    //  todo rank:113  dense_rank:112   row_number:123
    //  下面之所以需要减1是因为要符合题目，开头为0，因为排序函数开头都是1
    val r1 = order.join(detail, detail("order_id") === order("id"))
      .select("user_id", "sku_id")
      .distinct()
      .withColumn(
        "user_id",
        dense_rank() over (Window.orderBy("user_id"))
      )
      .withColumn("user_id", col("user_id") - 1)
      .withColumn(
        "sku_id",
        dense_rank() over (Window.orderBy("sku_id"))
      )
      .withColumn("sku_id", col("sku_id") - 1)

    r1.orderBy("user_id").limit(5).foreach{
      row => println(row(0)+":"+row(1))
    }


    //  todo 下面开始第二题

    //  todo 首先将sku_id的值全部换成sku_id+数字的形式
    val r2 = r1.withColumn(
        "sku_id",
        functions.concat(lit("sku_"), col("sku_id"))
      )
      .orderBy("user_id", "sku_id")

    // todo 然后将所有的sku_id排好序拿到收集到一个数组里面
    val sku_ids:Array[String] = r2.select("sku_id")
      .distinct()
      .orderBy(
        split(col("sku_id"), "id")(1).cast("int") // 根据sku_id字段里面的id进行切分
      )
      .map(
        x => x(0).toString
      )
      .collect()

    println("所有的sku_id为:")
//    sku_ids.foreach(println)


    /*
        todo 1.这里使用了povit将一列的值转化成新的列名(也就是把值变成了字段名)，然后sku_ids 是一个包含所有 sku_id 的列表，用于
         指定需要生成的列。
         2.然后搭配使用agg聚合函数，对分组后的数据进行聚合，如果user_id和sku_id都存在对应的值，则用1.0填充
         3.对于没有填充的则用0.0填充

         groupBy + pivot + agg + na.fill 是一种常见的固定搭配，用于实现类似于 One-Hot 编码 的效果。这种组合在 Spark 中
         非常实用，尤其是在需要对类别变量进行编码或生成矩阵的场景中。

         应用场景:
          1.One-Hot 编码： 将类别变量（如 sku_id）转换为二进制向量。
          2.生成矩阵：  生成用户-商品矩阵、用户-行为矩阵等。
          3.数据透视： 将长格式数据转换为宽格式数据。
     */
    val result=r2.groupBy("user_id")
      .pivot("sku_id",sku_ids)
      .agg(lit(1.0))
      .na.fill(0.0)
      .withColumn("user_id",col("user_id").cast(DoubleType))
      .orderBy("user_id")

    //  todo 1.result.columns.take(5):选择5个列    返回一个Array[String]
    //  todo 2.map(col):  是 Spark SQL 中的一个函数，用于将列名转换为列对象。
    //  todo 3.: _* 是 Scala 中的语法糖，用于将集合（如 Array、List）展开为可变参数。
    //  select 方法接受可变参数（Column*），因此需要使用 : _* 将数组展开为多个参数

    //  自我总结就是将取到的5个字段名从字符串转化为列对象，然后使用: _*展开，让select选择，因为seelct可以接受列对象
    println("第二题答案:")
    println(result.select(result.columns.take(5).map(col): _*).limit(1)
      .collect()(0).mkString(","))


    //  这里把结果存放到一个数据库里面去，为了方便给下面的推荐系统使用
    result.write.format("hive")
      .mode("overwrite")
      .saveAsTable("tzgc.skumat")

    result.show



    spark.close()

  }

}
